package chapter3

import org.apache.spark.{SparkConf, SparkContext}

/**
 * ./bin/spark-submit \
 * --class <main-class> \        # 应用程序主入口类
 * --master <master-url> \       # 集群的 Master Url
 * --deploy-mode <deploy-mode> \ # 部署模式
 * --conf <key>=<value> \        # 可选配置
 * ... # other options
 * <application-jar> \           # Jar 包路径
 * [application-arguments]       #传递给主入口类的参数
 *
 * <deploy-mode>
 * deploy-mode 有 cluster 和 client 两个可选参数，默认为 client。这里以 Spark On Yarn 模式对两者进行说明 ：
 *
 * 在 cluster 模式下，Spark Driver 在应用程序的 Master 进程内运行，该进程由群集上的 YARN 管理，提交作业的客户端可以在启动应用程序后关闭；
 * 在 client 模式下，Spark Driver 在提交作业的客户端进程中运行，Master 进程仅用于从 YARN 请求资源。
 *
 * <master-url>
 * local
 * 使用一个线程本地运行 Spark
 *
 * local[K]
 * 使用 K 个 worker 线程本地运行 Spark
 *
 * local[K,F]
 * 使用 K 个 worker 线程本地运行 , 第二个参数为 Task 的失败重试次数
 *
 * local[*]
 * 使用与 CPU 核心数一样的线程数在本地运行 Spark
 *
 * local[*,F]
 * 使用与 CPU 核心数一样的线程数在本地运行 Spark第二个参数为 Task 的失败重试次数
 *
 * local[x,y,z]
 * 伪集群模式（x为本机上启动的executor数，y为每个executor使用的core数，z为每个 executor使用的内存）
 *
 * spark://HOST:PORT
 * 连接至指定的 standalone 集群的 master 节点。端口号默认是 7077。
 *
 * spark://HOST1:PORT1,HOST2:PORT2
 * 如果 standalone 集群采用 Zookeeper 实现高可用，则必须包含由 zookeeper 设置的所有 master 主机地址。
 *
 * mesos://HOST:PORT
 * 连接至给定的 Mesos 集群。端口默认是 5050。对于使用了 ZooKeeper 的 Mesos cluster 来说，使用 mesos://zk://... 来指定地址，使用 --deploy-mode cluster 模式来提交。
 *
 * yarn
 * 连接至一个 YARN 集群，集群由配置的 HADOOP_CONF_DIR 或者 YARN_CONF_DIR 来决定。使用 --deploy-mode 参数来配置 client 或 cluster 模式。
 *
 * <h1>可选配置</h1>
 * 关于 Master 和 Woker 节点的所有可选配置如下，可以在 spark-env.sh 中进行对应的配置：
 * Environment Variable（环境变量）
 * Meaning（含义）
 *
 * SPARK_MASTER_HOST
 * master 节点地址
 *
 * SPARK_MASTER_PORT
 * master 节点地址端口（默认：7077）
 *
 * SPARK_MASTER_WEBUI_PORT
 * master 的 web UI 的端口（默认：8080）
 *
 * SPARK_MASTER_OPTS
 * 仅用于 master 的配置属性，格式是 "-Dx=y"（默认：none）,所有属性可以参考官方文档：spark-standalone-mode
 *
 * SPARK_LOCAL_DIRS
 * spark 的临时存储的目录，用于暂存 map 的输出和持久化存储 RDDs。多个目录用逗号分隔
 *
 * SPARK_WORKER_CORES
 * spark worker 节点可以使用 CPU Cores 的数量。（默认：全部可用）
 *
 * SPARK_WORKER_MEMORY
 * spark worker 节点可以使用的内存数量（默认：全部的内存减去 1GB）；
 *
 * SPARK_WORKER_PORT
 * spark worker 节点的端口（默认： random（随机））
 *
 * SPARK_WORKER_WEBUI_PORT
 * worker 的 web UI 的 Port（端口）（默认：8081）
 *
 * SPARK_WORKER_DIR
 * worker 运行应用程序的目录，这个目录中包含日志和暂存空间（default：SPARK_HOME/work）
 *
 * SPARK_WORKER_OPTS
 * 仅用于 worker 的配置属性，格式是 "-Dx=y"（默认：none）。所有属性可以参考官方文档：spark-standalone-mode
 *
 * SPARK_DAEMON_MEMORY
 * 分配给 spark master 和 worker 守护进程的内存。（默认： 1G）
 *
 * SPARK_DAEMON_JAVA_OPTS
 * spark master 和 worker 守护进程的 JVM 选项，格式是 "-Dx=y"（默认：none）
 *
 * SPARK_PUBLIC_DNS
 * spark master 和 worker 的公开 DNS 名称。（默认：none）
 *
 * 参数名	参数说明
 * --master	 master 的地址，提交任务到哪里执行，例如 spark://host:port,  yarn,  local
 * --deploy-mode	 在本地 (client) 启动 driver 或在 cluster 上启动，默认是 client
 * --class	 应用程序的主类，仅针对 java 或 scala 应用
 * --name	 应用程序的名称
 * --jars	 用逗号分隔的本地 jar 包，设置后，这些 jar 将包含在 driver 和 executor 的 classpath 下
 * --packages	 包含在driver 和executor 的 classpath 中的 jar 的 maven 坐标
 * --exclude-packages	 为了避免冲突 而指定不包含的 package
 * --repositories	 远程 repository
 * --conf PROP=VALUE  指定 spark 配置属性的值，例如 -conf spark.executor.extraJavaOptions="-XX:MaxPermSize=256m"
 * --properties-file	 加载的配置文件，默认为 conf/spark-defaults.conf
 * --driver-memory	 Driver内存，默认 1G
 * --driver-java-options	 传给 driver 的额外的 Java 选项
 * --driver-library-path	 传给 driver 的额外的库路径
 * --driver-class-path	 传给 driver 的额外的类路径
 * --driver-cores	 Driver 的核数，默认是1。在 yarn 或者 standalone 下使用
 * --executor-memory	 每个 executor 的内存，默认是1G
 * --total-executor-cores	 所有 executor 总共的核数。仅仅在 mesos 或者 standalone 下使用
 * --num-executors	 启动的 executor 数量。默认为2。在 yarn 下使用
 * --executor-core	 每个 executor 的核数。在yarn或者standalone下使用
 * <example>
 * spark-submit \
 * --class org.apache.spark.examples.SparkPi \
 * --master spark://hadoop001:7077 \
 * --executor-memory 2G \
 * --total-executor-cores 10 \
 * /usr/app/spark-2.4.0-bin-hadoop2.6/examples/jars/spark-examples_2.11-2.4.0.jar \
 * 100
 */
object Chapter3_3_2 {

  def main(args: Array[String]): Unit = {
    remote(args)
//    local(args)
  }

  /**
   * 本地运行
   *
   * @param args 程序运行参数
   */
  def local(args: Array[String]): Unit = {
    // 本地运行
    // 初始化SparkConf对象，设置基本任务参数
    val conf = new SparkConf()
      // 设置目标Master通信地址
      .setMaster("local")
      // 设置任务名称
      .setAppName("WC")
    computeRDD(conf)

  }

  /**
   * 远程运行
   *
   * @param args 程序运行参数
   */
  def remote(args: Array[String]): Unit = {
    val conf = new SparkConf()
      .setMaster("spark://localhost:7077")
      .setJars(List("P:\\Project\\Idea\\bigdata\\spark\\target\\spark-1.0-SNAPSHOT.jar")).set("class", "chapter3.Chapter3_3_2")
      .setAppName("WC")

    computeRDD(conf)

  }

  private def computeRDD(conf: SparkConf): Unit = {
    val sc = new SparkContext(conf)
    //读取文件并进行单词统计
    val hadoopRDD = sc.textFile("hdfs://localhost:9000/words.txt")
    val value = hadoopRDD.flatMap(_.split(" ")).map((_, 1))
    value.reduceByKey(_ + _, 1)
      .sortBy(_._2, ascending = false).top(1)
      .foreach(println(_))
    //停止sc,结束任务
    sc.stop()
  }
}
